package com.qianda.qdminiapp.common.future;

import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @ClassName DefaultCommandFutureManager
 * @Author cw
 * @Date 2019/8/1 16:58
 * @Version 1.0
 */
public class DefaultCommandFutureManager implements CommandFutureManager {
    private static final Logger LOG = LoggerFactory.getLogger(DefaultCommandFutureManager.class);

    private final Map<String, CommandFuture> futures;

    private final HashedWheelTimer timer;

    private final AtomicBoolean started;

    /**
     * 构造指令执行将来时持有
     */
    public DefaultCommandFutureManager() {
        this.futures = new ConcurrentHashMap<>(256);
        this.timer = new HashedWheelTimer(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "CommandFuturesCleaner");
            }
        }, 1, TimeUnit.SECONDS, 3600);
        this.started = new AtomicBoolean(true);
    }

    @Override
    public void start() throws Exception {
        LOG.info("Start command future manager");
        if (!started.compareAndSet(false, true)) {
            LOG.warn("Could not start command future manager, because started");
        }
    }

    @Override
    public void shutdown() throws Exception {
        if (started.getAndSet(false)) {
            LOG.info("Shutdown command future manager");
            timer.stop();
        } else {
            LOG.warn("Could not shutdown command future manager, because not started");
        }
    }

    @Override
    public boolean add(String futureId, CommandFuture future, long ttl) {
        if (!started.get()) {
            LOG.warn("Could not add command future, because not started");
            return false;
        }
        if (futures.putIfAbsent(futureId, future) != null) {
            LOG.warn("Could not add command future, because the future id " + futureId + " exists");
            return false;
        }
        timer.newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                if (futures.remove(futureId) != null) {
                    LOG.info("Clean command future: {}, because execute timeout", futureId);
                }
            }
        }, ttl, TimeUnit.SECONDS);

        return true;
    }

    @Override
    public void remove(String futureId) {
        futures.remove(futureId);
    }

    @Override
    public void setResult(String futureId, Object result) {
        CommandFuture future = futures.remove(futureId);
        if (future != null) {
            future.completed(result);
            LOG.info("Set command result: {}", result);
        } else {
            LOG.warn("Could not found that future: {}", result);
        }
    }

    @Override
    public void setException(String futureId, Exception ex) {
        CommandFuture future = futures.remove(futureId);
        if (future != null) {
            future.failed(ex);
            LOG.info("Set command exception: {}", futureId);
        } else {
            LOG.warn("Could not found that future: {}", futureId);
        }
    }

    @Override
    public void setCancellation(String futureId) {
        CommandFuture future = futures.remove(futureId);
        if (future != null) {
            future.cancel();
            LOG.info("Set command cancellation: {}", futureId);
        } else {
            LOG.warn("Could not found that future: {}", futureId);
        }
    }
}
